package com.glodon.paas.databag.biz.impl;

import java.util.HashMap;
import java.util.Map;

import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import org.springframework.web.client.RestClientException;
import org.springframework.web.client.RestTemplate;

import com.glodon.paas.databag.biz.DispatcherBiz;
import com.glodon.paas.databag.model.DatabagTask;
import com.glodon.paas.exception.PaasServerExceptionFactory;
import com.glodon.paas.job.JobConstants;
import com.glodon.paas.job.exception.JsonException;
import com.glodon.paas.job.util.JsonUtil;

@Component
public class DispatcherBizImpl implements DispatcherBiz {
    private final Logger logger = LoggerFactory.getLogger(DispatcherBizImpl.class);

    @Value("${job.task.url}")
    private String jobServerUrl;

    @Value("${databag.callback.url}")
    private String callback;
    
    private RestTemplate template = new RestTemplate();

    @Override
    public String dispatch(DatabagTask task) {
        String url = jobServerUrl.trim() + "/{type}";
        Map<String, String> params = prepareDispatcher(task);
        String json;
        try {
            json = template.postForObject(url, params, String.class, task.getType());
        } catch (RestClientException e) {
            throw PaasServerExceptionFactory.internalAPIException("submit task '" + task.getType() + "' failed.", e);
        }
        logger.info("send task '{}' to job server, response:{}", task.getType(), json);
        String taskId = parseTaskId(json);
        task.setTaskId(taskId);
        return taskId;
    }

    @SuppressWarnings({ "rawtypes", "unchecked" })
    private Map<String, String> prepareDispatcher(DatabagTask task) {
        Map params = new HashMap();
        params.put(JobConstants.USER_ID, task.getUserId());
        Map input = new HashMap();
        input.put(System.currentTimeMillis(), task.getInput());
        params.put(JobConstants.INPUT, input);
        params.put(JobConstants.OUTPUT, task.getOutput());
        params.put(JobConstants.PROPERTY_TASK_TYPE, task.getType());
        params.put(JobConstants.TASK_CALLBACK, getCallback());
        params.put(JobConstants.JOB_ID, task.getDatabagId());
        if ("dwg".equalsIgnoreCase(task.getType())) {
            params.put("online", true);
        }
        return params;
    }

    @SuppressWarnings("rawtypes")
    private String parseTaskId(String json) {
        Map result;
        try {
            result = JsonUtil.jsonToMap(json);
        } catch (JsonException e) {
            throw PaasServerExceptionFactory.validateException("bad json string from job-server", e);
        }
        String errorMsg = (String) result.get(JobConstants.ERROR_MSG);
        if (!StringUtils.isEmpty(errorMsg)) {
            throw PaasServerExceptionFactory.internalAPIException("job server Error:" + errorMsg, null);
        } else {
            return (String) result.get(JobConstants.PROPERTY_TASK_ID);
        }
    }

    public String getCallback() {
        return callback;
    }

    public void setCallback(String callback) {
        this.callback = callback;
    }

    public void setJobServerUrl(String jobServerUrl) {
        if (!jobServerUrl.endsWith("/")) {
            this.jobServerUrl = jobServerUrl + "/";
        } else {
            this.jobServerUrl = jobServerUrl;
        }
    }

    public void setTemplate(RestTemplate template) {
        this.template = template;
    }
}
